package com.neusoft.bigdata.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.neusoft.bigdata.hbase.util.CommonUtil;

/**
 * @author lihzh
 * @date 2013-3-27 下午4:18:10
 */
public class ForPrepareData extends CommonUtil {

	private static Configuration conf = null;
	
	// 2012年11月1日
	public static final long ENDTIME = CommonUtil.getTime(2012, 10, 1, 0, 0);

	public static final long INTERVAL = 1 * 60 * 1000L;

	// 批量插入的时间范围，单位次
	public static final long TIME_COUNT = 60 * 24 * 100; // 分钟 * 小时 * 天

	public static final String TABLE_NAME_METRIC = "metric";
	public static final String TABLE_NAME_METRIC_INDEX = "metric_time_index";
	public static final String TABLE_NAME_NEXT_TIME = "next_start_time";

	public static final String COLUMN_FAMILY_NAME_METRICS = "metrics";
	public static final String COLUMN_FAMILY_NAME_INFO = "info";
	public static final String COLUMN_FAMILY_NAME_ROWKEY = "rowkey";
	public static final String COLUMN_FAMILY_NAME_TIME = "time";
	public static final String COLUMN_NAME_ROWKEY = "rowkey";
	public static final String COLUMN_NAME_TIME = "time";

	public static final int FLUSH_COUNT = 10000;

	public static final String TIME_ROW_KEY = "nextTime";
	private static long tempStartTime = 0;

	private static HTable METRIC_HTABLE;
	private static HTable INDEX_HTABLE;
	private static HTable TIME_HTABLE;

	static {
		Configuration HBASE_CONFIG = new Configuration();
		HBASE_CONFIG.addResource("hbase-site.xml");
		conf = HBaseConfiguration.create(HBASE_CONFIG);
		try {
			METRIC_HTABLE = new HTable(conf, TABLE_NAME_METRIC);
			INDEX_HTABLE = new HTable(conf, TABLE_NAME_METRIC_INDEX);
			TIME_HTABLE = new HTable(conf, TABLE_NAME_NEXT_TIME);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * @param args
	 * @author lihzh
	 * @throws IOException
	 * @date 2013-3-27 下午4:18:10
	 */
	public static void main(String[] args) throws IOException {
		// 起始资源号
		int start = Integer.valueOf(args[0]);
		// 结束资源号
		int end = Integer.valueOf(args[1]);
		// 监控数据天数
		int dayCount = Integer.valueOf(args[2]);
		// 指标个数
		int metricCount = Integer.valueOf(args[3]);
		// 起始时间 0:从表获取 1: 初始化时间。其余，直接给定值。
		long startTime = Integer.valueOf(args[4]);
		long count = dayCount * 24 * 60;
		if (startTime == 0) {
			// 从数据获取下次监控指标的起始时间
			getNextStartTime();
		} else if (startTime == -1) {
			tempStartTime = STARTTIME;
		} else {
			tempStartTime = startTime;
		}
		// 向Metric表写数据
		getMetricPuts(start, end, tempStartTime, count, metricCount);
		// 向索引表写数据
		getIndexPuts(start, end, tempStartTime, count);
	}

	/**
	 * 获取监控数据起始时间
	 * 
	 * @author lihzh
	 * @throws IOException
	 * @date 2013-4-1 上午11:26:41
	 */
	private static void getNextStartTime() throws IOException {
		Get get = new Get(Bytes.toBytes(TIME_ROW_KEY));
		Result result = TIME_HTABLE.get(get);
		byte[] value = result.getValue(Bytes.toBytes(COLUMN_FAMILY_NAME_TIME),
				Bytes.toBytes(COLUMN_NAME_TIME));
		tempStartTime = value == null ? STARTTIME : Long.valueOf(new String(
				value));
	}

	/**
	 * 构造索引表的插入数据
	 * 
	 * @param start
	 * @param end
	 * @return
	 * @throws IOException
	 * @author lihzh
	 * @date 2013-4-1 上午10:57:49
	 */
	private static List<Put> getIndexPuts(int start, int end, long startTime,
			long recordTimes) throws IOException {
		List<Put> puts = new ArrayList<Put>(FLUSH_COUNT);
		int count = 0;
		for (int i = start; i <= end; i++) {
			String id = ID_PREFIX + CommonUtil.createId(i);
			long recordTime = startTime;
			for (int j = 0; j < recordTimes; j++) {
				String indexRowKey = recordTime + "_" + id;
				String rowKey = id + "_" + recordTime;
				recordTime += INTERVAL;
				Put put = new Put(Bytes.toBytes(indexRowKey));
				KeyValue kv = new KeyValue(Bytes.toBytes(indexRowKey),
						Bytes.toBytes(COLUMN_FAMILY_NAME_ROWKEY),
						Bytes.toBytes(COLUMN_NAME_ROWKEY),
						Bytes.toBytes(rowKey));
				put.add(kv);
				puts.add(put);
				count++;
				if (count == FLUSH_COUNT) {
					addRecords(INDEX_HTABLE, puts);
					count = 0;
					puts.clear();
				}
			}
		}
		if (puts.size() > 0) {
			addRecords(INDEX_HTABLE, puts);
		}
		return puts;
	}

	public static List<Put> getMetricPuts(int start, int end, long startTime,
			long recordTimes, int metricCount) throws IOException {
		List<Put> puts = new ArrayList<Put>(FLUSH_COUNT);
		int count = 0;
		long recordTime = startTime;
		for (int i = start; i <= end; i++) {
			String id = ID_PREFIX + CommonUtil.createId(i);
			recordTime = startTime;
			for (int j = 0; j < recordTimes; j++) {
				String rowKey = id + "_" + recordTime;
				recordTime += INTERVAL;
				Put put = new Put(Bytes.toBytes(rowKey));
				// 构造Metrics列族, 100指标列
				for (int l = 1; l <= metricCount; l++) {
					KeyValue kv = new KeyValue(Bytes.toBytes(rowKey),
							Bytes.toBytes(COLUMN_FAMILY_NAME_METRICS),
							Bytes.toBytes(METRIC_NAME_PREFIX + l),
							Bytes.toBytes(getValue()));
					put.add(kv);
				}
				// 构造info列族，rowkey列
				KeyValue infoKVKey = new KeyValue(Bytes.toBytes(rowKey),
						Bytes.toBytes(COLUMN_FAMILY_NAME_INFO),
						Bytes.toBytes(COLUMN_NAME_ROWKEY), Bytes.toBytes(id));
				// 构造info列族, time列
				KeyValue infoKVTime = new KeyValue(Bytes.toBytes(rowKey),
						Bytes.toBytes(COLUMN_FAMILY_NAME_INFO),
						Bytes.toBytes(COLUMN_NAME_TIME), Bytes.toBytes(id));
				put.add(infoKVKey);
				put.add(infoKVTime);
				puts.add(put);
				count++;
				if (count == FLUSH_COUNT) {
					addRecords(METRIC_HTABLE, puts);
					count = 0;
					puts.clear();
				}
			}
		}
		if (puts.size() > 0) {
			addRecords(METRIC_HTABLE, puts);
		}
		saveNextRecordTime(recordTime);
		return puts;
	}

	private static void saveNextRecordTime(long currentTime) throws IOException {
		Put put = new Put(Bytes.toBytes(TIME_ROW_KEY));
		KeyValue kv = new KeyValue(Bytes.toBytes(TIME_ROW_KEY),
				Bytes.toBytes(COLUMN_FAMILY_NAME_TIME),
				Bytes.toBytes(COLUMN_NAME_TIME), Bytes.toBytes(String
						.valueOf(currentTime + INTERVAL)));
		put.add(kv);
		TIME_HTABLE.put(put);
	}

	public static void addRecords(HTable table, List<Put> puts) {
		try {
			long startTime = System.currentTimeMillis();
			table.put(puts);
			long endTime = System.currentTimeMillis();
			System.out.println("insert recored count: " + puts.size()
					+ " into table " + new String(table.getTableName()));
			System.out.println("Insert data cost time: "
					+ (endTime - startTime));
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

}
